package com.tlgen.coopera.service.impl;

import com.tlgen.coopera.model.BeatRegistryServer;
import com.tlgen.coopera.model.RegistryServer;
import com.tlgen.coopera.factory.RegistryLocalFactory;
import com.tlgen.coopera.runner.MyDelayTask;
import com.tlgen.coopera.service.HeartBeatService;
import com.tlgen.coopera.service.RegistryService;
import com.tlgen.coopera.utils.DateUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.DelayQueue;
import java.util.stream.Collectors;

/*
 *@title HeartBeatServiceImpl
 *@description
 *@author ta151486
 *@version 1.0
 *@create 2023/12/13 15:14
 */
@Service
public class HeartBeatServiceImpl implements HeartBeatService {

    @Autowired
    private Environment environment;

    @Autowired
    private RegistryService registryService;

    @Autowired
    private DelayQueue<MyDelayTask> delayQueue;

    @Override
    public void sendBeat() {
        RegistryLocalFactory instance = RegistryLocalFactory.getInstance();
        List<RegistryServer> registryList = instance.getRegistryList();
        List<BeatRegistryServer> beatList = registryList
                .stream().map(x -> {
                    BeatRegistryServer beatRegistryServer = new BeatRegistryServer();
                    BeanUtils.copyProperties(x, beatRegistryServer);
                    beatRegistryServer.setSendBeatTime(DateUtils.LocalDateTimeToString(LocalDateTime.now()));
                    return beatRegistryServer;
                }).collect(Collectors.toList());
        instance.setBeatList(beatList);
        Integer localPort = Integer.valueOf(Optional.ofNullable(environment.getProperty("server.port"))
                .orElse("8080"));
        for (BeatRegistryServer beatRegistryServer : beatList) {
            String serverIp = beatRegistryServer.getIp(); // 设置目标服务器的IP地址
            int serverPort = beatRegistryServer.getPort(); // 设置目标服务器的端口号
            // 发送心跳之前, 定义延迟任务, 开始计时, 检测出非健康节点, 超过 6 秒接收不到 ACK 则被认定为非健康节点
            delayQueue.add(new MyDelayTask(beatRegistryServer, 6 * 1000));
            try (DatagramSocket socket = new DatagramSocket()) {
                InetAddress address = InetAddress.getByName(serverIp);
                String data = "beat:" + beatRegistryServer.getIp() + ":" + localPort; // beat:192.168.1.5:9994
                byte[] buffer = data.getBytes(StandardCharsets.UTF_8); // 要发送的数据内容
                DatagramPacket packet = new DatagramPacket(buffer, buffer.length, address, serverPort);
                socket.send(packet); // 发送心跳包到目标服务器
                socket.close();
                System.out.println("已向 " + serverIp + ":" + serverPort + " 发送心跳包！");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void receive() {
        // 接收 udp 心跳
        try {
            Integer localPort = Integer.valueOf(Optional.ofNullable(environment.getProperty("server.port"))
                    .orElse("8080"));
            DatagramSocket socket = new DatagramSocket(localPort); // 创建监听 UDP 数据的socket，并绑定端口号
            byte[] buffer = new byte[1024]; // 创建缓冲区用于存储接收到的数据
            DatagramPacket packet = new DatagramPacket(buffer, buffer.length); // 创建数据包用于接收UDP数据
            while (true) {
                socket.receive(packet); // 接收 UDP 数据，此方法会阻塞直到接收到数据为止
                String receivedData = new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); // 从数据包中获取接收到的数据
                String[] split = receivedData.split(":");
                System.out.println("接收到来自 " + split[1] + ":" + split[2] + " 回传的 ACK 确认");
                System.out.println(receivedData);
                packet.setLength(buffer.length); // 重置数据包的长度，以便下一次接收数据
                if (receivedData.startsWith("ACK")) {
                    // 收到 ACK, 更新服务列表中健康服务的心跳时间
                    List<BeatRegistryServer> beatList = RegistryLocalFactory.getInstance().getBeatList();
                    List<BeatRegistryServer> collect = beatList.stream().peek
                            (x -> x.setReceiveAckTime(DateUtils.LocalDateTimeToString(LocalDateTime.now()))).collect(Collectors.toList());
                    RegistryLocalFactory.getInstance().setBeatList(collect);
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

}
